package com.kalua.list.dbaccess.datamgmt;

import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.set;
import static com.kalua.list.dbaccess.datamgmt.SequenceRowMapper.DEFAULT_INCREMENT;
import static com.kalua.list.dbaccess.datamgmt.SequenceRowMapper.NAME_COL_NAME;
import static com.kalua.list.dbaccess.datamgmt.SequenceRowMapper.SEQUENCE_COL_FAMILY_NAME;
import static com.kalua.list.dbaccess.datamgmt.SequenceRowMapper.VALUE_COL_NAME;
import static com.obixlabs.commons.util.LockUtil.acquireLock;
import static com.obixlabs.commons.util.LockUtil.releaseLock;

import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.springframework.data.cassandra.core.CassandraTemplate;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.google.common.collect.Maps;
import com.kalua.list.domainmodel.SequenceValue;
import com.kalua.list.domainmodel.SequenceValueImpl;

public class SequenceDAOImpl implements SequenceDAO {	
	
	private static final int DEFAULT_INITIAL_VALUE = 0;
	
	private CassandraTemplate template;
	private ConcurrentMap<String, Lock> sequenceLocks;
				
	public SequenceDAOImpl(CassandraTemplate template) {
		this.sequenceLocks = Maps.newConcurrentMap();
		setTemplate(template);
	}
	
	@Override
	public SequenceValue getNext(String sequenceId) {
		SequenceValue result;
		Lock sequenceLock = getLockForSequence(sequenceId);
		acquireLock(sequenceLock);
		try {
			Statement statement = buildReadStatement(sequenceId);
			ResultSet resultSet = getTemplate().getSession().execute(statement);
			List<SequenceValue> results = 
							getTemplate().process(	resultSet, 
																new SequenceRowMapper());
			if (!results.isEmpty())
				result = results.get(0);
			else result = new SequenceValueImpl(
											DEFAULT_INITIAL_VALUE, 
											DEFAULT_INITIAL_VALUE + DEFAULT_INCREMENT);
			
			updateToNextValue(result, sequenceId);			
			
		} finally {
			releaseLock(sequenceLock);
		}
		return result;
	}
	
	private void updateToNextValue(SequenceValue value, String sequenceId) {
		Statement statement = 
					QueryBuilder.update(SEQUENCE_COL_FAMILY_NAME)
		        						.with(set(VALUE_COL_NAME, value.getCurrentValue()))
		        						.where(eq(NAME_COL_NAME, sequenceId));
		getTemplate().getSession().execute(statement);		
	}

	public CassandraTemplate getTemplate() { return this.template; }
	private void setTemplate(CassandraTemplate template) 
	{ this.template = template; }	

	protected Statement buildReadStatement(String sequenceId) {
		Statement statement = QueryBuilder.select().all()
		        							.from(SEQUENCE_COL_FAMILY_NAME)
		        							.where(eq(NAME_COL_NAME, sequenceId));
		return statement;
	}	
	
	private Lock getLockForSequence(String sequenceId) {
		Lock result;		
		Lock newLock = new ReentrantLock();
		result = this.sequenceLocks.putIfAbsent(sequenceId, newLock);
		if (result == null)
			result = newLock;		
		return result;
	}
}